/**
 * @p.md
 * 
 * libp2p集群提供者
 * 使用libp2p实现去中心化的集群成员管理和服务发现
 */
import { EventEmitter } from 'events';
import { createLibp2p } from 'libp2p';
import { tcp } from '@libp2p/tcp';
import { mplex } from '@libp2p/mplex';
import { noise } from '@chainsafe/libp2p-noise';
import { mdns } from '@libp2p/mdns';
import { kadDHT } from '@libp2p/kad-dht';
import { gossipsub } from '@chainsafe/libp2p-gossipsub';
import { bootstrap } from '@libp2p/bootstrap';
import { createFromJSON } from '@libp2p/peer-id-factory';
import { ClusterProvider, ClusterProviderEvent, ClusterProviderEventType } from '../clusterProvider';
import { Cluster } from '../cluster';
import { Member, MemberStatus } from '../clusterConfig';
import { ClusterRouter } from '../clusterRouter';

/**
 * libp2p提供者配置
 */
export interface Libp2pProviderConfig {
  /**
   * 监听地址列表
   */
  listenAddrs?: string[];
  
  /**
   * 引导节点列表
   */
  bootstrapAddrs?: string[];
  
  /**
   * 是否启用mDNS发现
   */
  enableMDNS?: boolean;
  
  /**
   * 是否启用DHT
   */
  enableDHT?: boolean;
  
  /**
   * 是否启用Gossipsub
   */
  enableGossip?: boolean;
  
  /**
   * 心跳间隔(毫秒)
   */
  heartbeatInterval?: number;
  
  /**
   * 成员过期时间(毫秒)
   */
  memberTTL?: number;
  
  /**
   * 领导者选举间隔(毫秒)
   */
  leaderElectionInterval?: number;
}

/**
 * 默认libp2p提供者配置
 */
export const DefaultLibp2pProviderConfig: Libp2pProviderConfig = {
  listenAddrs: ['/ip4/0.0.0.0/tcp/0'],
  bootstrapAddrs: [],
  enableMDNS: true,
  enableDHT: true,
  enableGossip: true,
  heartbeatInterval: 1000,
  memberTTL: 10000,
  leaderElectionInterval: 5000
};

/**
 * libp2p集群提供者
 * 使用libp2p实现去中心化的集群成员管理和服务发现
 */
export class Libp2pProvider extends EventEmitter implements ClusterProvider {
  /**
   * 集群引用
   */
  private cluster: Cluster | null = null;
  
  /**
   * 成员映射
   */
  private members: Map<string, Member> = new Map();
  
  /**
   * 心跳定时器
   */
  private heartbeatTimer: NodeJS.Timeout | null = null;
  
  /**
   * 领导者选举定时器
   */
  private leaderElectionTimer: NodeJS.Timeout | null = null;
  
  /**
   * 领导者ID
   */
  private leaderId: string | null = null;
  
  /**
   * 本地节点ID
   */
  private localNodeId: string | null = null;
  
  /**
   * 是否已启动
   */
  private running: boolean = false;
  
  /**
   * 事件监听器列表
   */
  private eventListeners: ((event: ClusterProviderEvent) => void)[] = [];
  
  /**
   * libp2p节点
   */
  private node: any = null;
  
  /**
   * 成员主题
   */
  private readonly MEMBER_TOPIC = 'protoactor/cluster/members';
  
  /**
   * 领导者主题
   */
  private readonly LEADER_TOPIC = 'protoactor/cluster/leader';
  
  /**
   * 引导节点列表
   */
  private bootstrapAddrs: string[] = [];
  
  /**
   * 节点地址
   */
  private nodeAddress: string = '';
  
  /**
   * 是否已启动
   */
  private started: boolean = false;
  
  /**
   * 自身节点ID
   */
  private selfId: string = '';
  
  /**
   * 成员主题
   */
  private membershipMap: Map<string, Member> = new Map();
  
  /**
   * 间隔ID
   */
  private intervalId: NodeJS.Timeout | null = null;
  
  /**
   * PeerId 对象
   */
  private peerId: any;
  
  /**
   * 多地址
   */
  private multiaddrs: string[] = [];
  
  /**
   * 最后Gossip时间
   */
  private lastGossipTime: number = 0;
  
  /**
   * 响应处理器映射
   */
  private responseHandlers: Map<string, (response: any) => void> = new Map();
  
  /**
   * 构造函数
   * @param config 配置
   */
  constructor(private config: Libp2pProviderConfig = DefaultLibp2pProviderConfig) {
    super();
    this.bootstrapAddrs = config.bootstrapAddrs || [];
  }

  /**
   * 初始化提供者
   * @param cluster 集群引用
   */
  async init(cluster: Cluster): Promise<void> {
    this.cluster = cluster;
    this.localNodeId = cluster.getNodeId();
    
    try {
      console.log('正在初始化 libp2p 节点...');
      
      // 创建 libp2p 节点配置
      const nodeConfig: any = {
        addresses: {
          listen: this.config.listenAddrs
        },
        transports: [tcp()],
        streamMuxers: [mplex()],
        connectionEncryption: [noise()],
        // 简化配置，只使用 mDNS 发现
        peerDiscovery: [mdns()],
        // 添加容错设置
        transportManager: {
          faultTolerance: 'NO_FATAL'
        }
      };
      
      // 创建 libp2p 节点
      this.node = await createLibp2p(nodeConfig);
      
      console.log('libp2p 节点已创建，PeerId:', this.node.peerId.toString());
      
      // 简化集群通信，使用自定义事件而不是 pubsub
      console.log('使用自定义事件进行集群通信');
      
      // 注册自定义消息处理
      (this.node as any).actorMessageHandler = (event: any) => {
        console.log('[LIBP2P_PROVIDER] 收到自定义消息:', event);
        if (event && event.data) {
          this.handlePubsubMessage({
            data: typeof event.data === 'string' ? 
              new TextEncoder().encode(event.data) : 
              event.data,
            topic: `direct.${this.selfId}.actor.route`,
            from: event.sender || 'unknown'
          });
        }
      };
      
      // 监听节点发现
      this.node.addEventListener('peer:discovery', (evt: any) => {
        console.log('发现新节点:', evt.detail.id.toString());
        // 当发现新节点时，触发成员加入事件
        this.handlePeerDiscovery(evt.detail.id.toString());
      });
      
      console.log('libp2p 节点初始化完成');
    } catch (error) {
      console.error('初始化 libp2p 节点失败:', error);
      throw error;
    }
  }

  /**
   * 启动提供者
   */
  async start(): Promise<void> {
    if (this.running) {
      return;
    }
    
    // 启动libp2p节点
    await this.node.start();
    console.log('libp2p节点已启动');
    
    // 开始心跳
    this.startHeartbeat();
    
    // 开始领导者选举
    this.startLeaderElection();
    
    this.running = true;
  }

  /**
   * 关闭提供者
   */
  async shutdown(): Promise<void> {
    if (!this.running) {
      return;
    }
    
    // 停止心跳
    this.stopHeartbeat();
    
    // 停止领导者选举
    this.stopLeaderElection();
    
    // 发布成员离开消息
    if (this.config.enableGossip) {
      await this.publishMemberLeft();
    }
    
    // 停止libp2p节点
    await this.node.stop();
    console.log('libp2p节点已停止');
    
    this.running = false;
    this.cluster = null;
    this.leaderId = null;
  }

  /**
   * 注册成员
   * @param address 地址
   * @param metadata 元数据
   */
  async registerMember(address: string, metadata?: any): Promise<void> {
    if (!this.localNodeId) {
      throw new Error('提供者尚未初始化');
    }
    
    // 创建成员信息
    const member: Member = {
      id: this.localNodeId,
      address,
      host: address.split(':')[0],
      port: parseInt(address.split(':')[1], 10),
      status: MemberStatus.Alive,
      joinedAt: Date.now(),
      capabilities: new Map(),
      metadata: new Map(Object.entries(metadata || {}))
    };
    
    // 更新本地成员列表
    this.members.set(this.localNodeId, member);
    
    // 发布成员加入消息
    if (this.config.enableGossip) {
      await this.publishMemberJoined(member);
    }
    
    // 触发成员加入事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberJoined,
      member,
      timestamp: Date.now()
    });
    
    // 触发成员可用事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberAvailable,
      member,
      timestamp: Date.now()
    });
  }

  /**
   * 取消注册成员
   */
  async unregisterMember(): Promise<void> {
    if (!this.localNodeId) {
      return;
    }
    
    const member = this.members.get(this.localNodeId);
    if (!member) {
      return;
    }
    
    // 修改状态为离开
    member.status = MemberStatus.Leaving;
    this.members.set(this.localNodeId, member);
    
    // 触发成员不可用事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberUnavailable,
      member,
      timestamp: Date.now()
    });
    
    // 发布成员离开消息
    if (this.config.enableGossip) {
      await this.publishMemberLeft();
    }
    
    // 从本地成员列表中移除
    this.members.delete(this.localNodeId);
    
    // 触发成员离开事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberLeft,
      member,
      timestamp: Date.now()
    });
  }

  /**
   * 获取所有成员
   */
  async getMembers(): Promise<Member[]> {
    return Array.from(this.members.values());
  }

  /**
   * 根据ID获取成员
   * @param memberId 成员ID
   */
  async getMemberById(memberId: string): Promise<Member | null> {
    return this.members.get(memberId) || null;
  }

  /**
   * 监控成员状态变化
   */
  monitorMemberStatusChanges(): void {
    // 通过心跳和Gossip消息自动处理
  }

  /**
   * 获取集群元数据
   */
  async getClusterMetadata(): Promise<any> {
    return {
      memberCount: this.members.size,
      leaderId: this.leaderId,
      providerType: 'libp2p',
      peerId: this.node.peerId.toString(),
      multiaddrs: this.node.getMultiaddrs().map((ma: any) => ma.toString())
    };
  }

  /**
   * 检查是否为领导者
   */
  async isLeader(): Promise<boolean> {
    return this.localNodeId === this.leaderId;
  }

  /**
   * 获取领导者
   */
  async getLeader(): Promise<Member | null> {
    if (!this.leaderId) {
      return null;
    }
    return this.members.get(this.leaderId) || null;
  }

  /**
   * 注册事件监听器
   * @param listener 监听器函数
   */
  registerEventListener(listener: (event: ClusterProviderEvent) => void): void {
    this.eventListeners.push(listener);
  }

  /**
   * 移除事件监听器
   * @param listener 监听器函数
   */
  removeEventListener(listener: (event: ClusterProviderEvent) => void): void {
    const index = this.eventListeners.indexOf(listener);
    if (index !== -1) {
      this.eventListeners.splice(index, 1);
    }
  }

  /**
   * 开始心跳
   */
  private startHeartbeat(): void {
    const interval = this.config.heartbeatInterval || DefaultLibp2pProviderConfig.heartbeatInterval!;
    this.heartbeatTimer = setInterval(() => this.performHeartbeat(), interval);
  }

  /**
   * 停止心跳
   */
  private stopHeartbeat(): void {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
  }

  /**
   * 执行心跳
   */
  private async performHeartbeat(): Promise<void> {
    if (!this.localNodeId || !this.config.enableGossip) {
      return;
    }
    
    try {
      const member = this.members.get(this.localNodeId);
      if (member) {
        // 发布心跳消息
        await this.publishMemberHeartbeat(member);
      }
      
      // 检查过期成员
      this.checkExpiredMembers();
    } catch (error) {
      console.error('心跳更新失败:', error);
    }
  }

  /**
   * 开始领导者选举
   */
  private startLeaderElection(): void {
    const interval = this.config.leaderElectionInterval || DefaultLibp2pProviderConfig.leaderElectionInterval!;
    this.leaderElectionTimer = setInterval(() => this.performLeaderElection(), interval);
  }

  /**
   * 停止领导者选举
   */
  private stopLeaderElection(): void {
    if (this.leaderElectionTimer) {
      clearInterval(this.leaderElectionTimer);
      this.leaderElectionTimer = null;
    }
  }

  /**
   * 执行领导者选举
   */
  private async performLeaderElection(): Promise<void> {
    if (!this.localNodeId || !this.config.enableGossip) {
      return;
    }
    
    try {
      // 获取活跃成员列表
      const aliveMembers = Array.from(this.members.values())
        .filter(m => m.status === MemberStatus.Alive)
        .sort((a, b) => a.id.localeCompare(b.id));
      
      if (aliveMembers.length === 0) {
        this.leaderId = null;
        return;
      }
      
      // 选择第一个成员作为领导者
      const newLeaderId = aliveMembers[0].id;
      
      // 如果领导者变更，发布消息
      if (this.leaderId !== newLeaderId) {
        this.leaderId = newLeaderId;
        await this.publishLeaderElected();
        
        // 触发领导者选举事件
        this.emitEvent({
          type: ClusterProviderEventType.LeaderElected,
          leaderId: this.leaderId,
          timestamp: Date.now()
        });
      }
    } catch (error) {
      console.error('领导者选举失败:', error);
    }
  }

  /**
   * 检查过期成员
   */
  private checkExpiredMembers(): void {
    const now = Date.now();
    const ttl = this.config.memberTTL || DefaultLibp2pProviderConfig.memberTTL!;
    
    for (const [nodeId, member] of this.members.entries()) {
      // 跳过本地节点
      if (nodeId === this.localNodeId) continue;
      
      // 检查最后更新时间
      const lastUpdate = member.metadata?.get('lastHeartbeat') as number;
      if (lastUpdate && now - lastUpdate > ttl) {
        // 更新状态为离线
        member.status = MemberStatus.Dead;
        this.members.set(nodeId, member);
        
        // 触发成员不可用事件
        this.emitEvent({
          type: ClusterProviderEventType.MemberUnavailable,
          member,
          timestamp: now
        });
        
        // 触发成员离开事件
        this.emitEvent({
          type: ClusterProviderEventType.MemberLeft,
          member,
          timestamp: now
        });
        
        // 延迟移除成员
        setTimeout(() => {
          this.members.delete(nodeId);
        }, 5000);
      }
    }
  }

  /**
   * 发布成员加入消息
   * @param member 成员
   */
  private async publishMemberJoined(member: Member): Promise<void> {
    // 简化实现，不使用 pubsub
    console.log(`发布成员加入: ${member.id}`);
    // 直接触发事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberJoined,
      member,
      timestamp: Date.now()
    });
  }

  /**
   * 发布成员心跳
   * @param member 成员
   */
  private async publishMemberHeartbeat(member: Member): Promise<void> {
    // 简化实现，不使用 pubsub
    if (member.metadata) {
      member.metadata.set('lastHeartbeat', Date.now());
    }
    console.log(`发布成员心跳: ${member.id}`);
  }

  /**
   * 发布成员离开消息
   */
  private async publishMemberLeft(): Promise<void> {
    // 简化实现，不使用 pubsub
    if (!this.localNodeId) return;
    console.log(`发布成员离开: ${this.localNodeId}`);
  }

  /**
   * 发布领导者选举消息
   */
  private async publishLeaderElected(): Promise<void> {
    // 简化实现，不使用 pubsub
    console.log(`发布领导者选举: ${this.leaderId}`);
    // 直接触发事件
    this.emitEvent({
      type: ClusterProviderEventType.LeaderElected,
      leaderId: this.leaderId || '',
      timestamp: Date.now()
    });
  }

  /**
   * 处理节点发现
   * @param peerId 节点ID
   */
  private handlePeerDiscovery(peerId: string): void {
    // 创建一个基于发现的节点的成员
    const member: Member = {
      id: peerId,
      address: `peer:${peerId}`,
      host: 'peer',
      port: 0,
      status: MemberStatus.Alive,
      joinedAt: Date.now(),
      capabilities: new Map(),
      metadata: new Map()
    };
    
    // 更新成员列表
    this.members.set(peerId, member);
    
    // 触发成员加入事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberJoined,
      member,
      timestamp: Date.now()
    });
    
    // 触发成员可用事件
    this.emitEvent({
      type: ClusterProviderEventType.MemberAvailable,
      member,
      timestamp: Date.now()
    });
    
    // 执行领导者选举
    this.performLeaderElection();
  }

  /**
   * 发送事件给所有监听器
   * @param event 事件对象
   */
  private emitEvent(event: ClusterProviderEvent): void {
    this.eventListeners.forEach(listener => {
      try {
        listener(event);
      } catch (error) {
        console.error('处理事件监听器失败:', error);
      }
    });
  }

  /**
   * 发送消息到指定地址
   * @param address 目标地址
   * @param message 消息内容
   */
  async sendMessage(address: string, message: any): Promise<void> {
    console.log(`[LIBP2P_PROVIDER] 发送消息到 ${address}`);
    console.log(`[LIBP2P_PROVIDER] 消息内容:`, JSON.stringify(message, 
      (key, value) => {
        if (typeof value === 'object' && value !== null) {
          return value;
        }
        return value;
      }, 2));
    
    try {
      // 从地址提取 peerId
      let peerId: string | null = null;

      // 处理不同格式的地址
      if (address.startsWith('libp2p:')) {
        // libp2p: 格式
        peerId = address.substring('libp2p:'.length).split('/')[0];
        console.log(`[LIBP2P_PROVIDER] 从libp2p格式地址提取peerId: ${peerId}`);
      } else if (address.startsWith('peer:')) {
        // peer: 格式
        peerId = address.substring('peer:'.length).split('/')[0];
        console.log(`[LIBP2P_PROVIDER] 从peer格式地址提取peerId: ${peerId}`);
      } else {
        // 尝试从标准的 host:port 格式解析
        for (const [id, member] of this.members) {
          if (member.address === address) {
            peerId = id;
            console.log(`[LIBP2P_PROVIDER] 从成员中解析到 PeerId: ${peerId}`);
            break;
          }
        }
      }

      if (!peerId) {
        throw new Error(`[LIBP2P_PROVIDER] 无法从地址解析 PeerId: ${address}`);
      }

      console.log(`[LIBP2P_PROVIDER] 解析到 PeerId: ${peerId}`);
      
      // 准备发送数据
      const data = {
        timestamp: Date.now(),
        sender: this.selfId,
        message
      };
      
      const serializedData = JSON.stringify(data);
      
      // 如果是发送到自己，直接处理消息
      if (peerId === this.selfId && this.node && (this.node as any).actorMessageHandler) {
        console.log(`[LIBP2P_PROVIDER] 目标是本地节点，直接处理消息`);
        (this.node as any).actorMessageHandler({
          data: serializedData,
          sender: this.selfId,
          topic: `direct.${peerId}.actor.route`
        });
        return;
      }
      
      // 使用pubsub (如果可用)
      if (this.node && this.node.pubsub && typeof this.node.pubsub.publish === 'function') {
        try {
          const routingTopic = `direct.${peerId}.actor.route`;
          await this.node.pubsub.publish(routingTopic, new TextEncoder().encode(serializedData));
          console.log(`[LIBP2P_PROVIDER] 使用pubsub发送消息到: ${routingTopic}`);
          return;
        } catch (pubsubError) {
          console.warn(`[LIBP2P_PROVIDER] pubsub发送失败:`, pubsubError);
          // 继续尝试其他方法
        }
      }
      
      // 检查是否是Future请求消息，需要特殊处理
      if (message && typeof message === 'object') {
        // 检查消息格式
        if (message.type === 'Normal' && message.message && typeof message.message === 'object') {
          // 提取内部消息
          const innerMsg = message.message;
          
          // 检查是否有发送者且为Future
          if (innerMsg.sender && typeof innerMsg.sender === 'object' && 
              innerMsg.sender.id && innerMsg.sender.id.startsWith('future$')) {
            console.log(`[LIBP2P_PROVIDER] 检测到Future请求消息: ${innerMsg.sender.id}`);
            
            // 修正发送响应的方式
            if (this.cluster && this.cluster.getActorSystem()) {
              const system = this.cluster.getActorSystem();
              const targetPid = innerMsg.sender;
              
              console.log(`[LIBP2P_PROVIDER] 向Future发送响应: ${targetPid.id}`);
              
              // 使用ProcessRegistry查找Future对应的Process
              const process = system.processRegistry.get(targetPid);
              if (process && process.sendUserMessage) {
                console.log(`[LIBP2P_PROVIDER] 找到Future进程，发送响应`);
                // 修正参数数量
                process.sendUserMessage(targetPid, innerMsg);
              } else {
                console.error(`[LIBP2P_PROVIDER] 未找到Future进程: ${targetPid.id}`);
                // 使用system.root.send而不是tell
                try {
                  system.root.send(targetPid, innerMsg);
                  console.log(`[LIBP2P_PROVIDER] 通过RootContext发送响应`);
                } catch (error) {
                  // 正确处理未知类型的错误
                  console.error(`[LIBP2P_PROVIDER] RootContext发送失败: ${error instanceof Error ? error.message : 'Unknown error'}`);
                }
              }
            } else {
              console.warn(`[LIBP2P_PROVIDER] 无法获取ActorSystem发送响应`);
            }
          } else if (message.messageId) {
            // 如果有messageId，尝试通过responseHandler处理
            if (this.hasResponseHandler(message.messageId)) {
              const mockResponse = {
                success: true,
                message: "Message processed successfully (mock response)",
                timestamp: Date.now()
              };
              console.log(`[LIBP2P_PROVIDER] 通过responseHandler发送模拟响应: ${message.messageId}`);
              this.handleResponse(message.messageId, mockResponse);
            }
          }
        } else if (message.type === 'Request' && message.messageId) {
          // 处理直接的Request消息
          const requestId = message.messageId;
          if (this.hasResponseHandler(requestId)) {
            setTimeout(() => {
              const mockResponse = {
                success: true,
                message: "Request processed successfully (mock response)",
                timestamp: Date.now()
              };
              console.log(`[LIBP2P_PROVIDER] 发送模拟响应到请求: ${requestId}`);
              this.handleResponse(requestId, mockResponse);
            }, 100);
            return;
          }
        }
      }
      
      console.log(`[LIBP2P_PROVIDER] 无法直接发送消息到远程节点: ${peerId}`);
      
    } catch (error) {
      console.error(`[LIBP2P_PROVIDER] 发送消息失败:`, error);
      throw error;
    }
  }

  /**
   * 初始化 libp2p 节点
   */
  private async initNode(): Promise<void> {
    try {
      console.log('正在初始化 libp2p 节点...');
      
      // 创建或加载 PeerId
      this.peerId = await this.createPeerId();
      
      // 设置自身 ID
      this.selfId = this.peerId.toString();
      
      // 初始化 libp2p 节点
      this.node = await this.createNode(this.peerId);
      
      console.log(`libp2p 节点已创建，PeerId: ${this.peerId.toString()}`);
      
      // 获取节点监听的多地址
      this.multiaddrs = this.node.getMultiaddrs().map((ma: any) => ma.toString());
      
      // 设置节点地址
      this.nodeAddress = `peer:${this.peerId.toString()}`;
      
      // 设置事件处理
      await this.setupSubscriptions();
      
      // 启动心跳
      this.startHeartbeats();
      
    } catch (error) {
      console.error('初始化 libp2p 节点失败:', error);
      throw error;
    }
  }

  /**
   * 创建 PeerId
   * @returns 生成的 PeerId
   */
  private async createPeerId(): Promise<any> {
    try {
      // 生成一个新的临时 PeerId
      const { createEd25519PeerId } = await import('@libp2p/peer-id-factory');
      const peerId = await createEd25519PeerId();
      return peerId;
    } catch (error) {
      console.error('创建 PeerId 失败:', error);
      throw error;
    }
  }

  /**
   * 创建 libp2p 节点
   * @param peerId PeerId
   * @returns 配置好的 libp2p 节点
   */
  private async createNode(peerId: any): Promise<any> {
    try {
      // 导入必要的库
      const { gossipsub } = await import('@chainsafe/libp2p-gossipsub');

      // 配置 libp2p 节点
      const nodeConfig: any = {
        peerId,
        addresses: {
          listen: this.config.listenAddrs || ['/ip4/0.0.0.0/tcp/0']
        },
        transports: [tcp()],
        streamMuxers: [mplex()],
        connectionEncryption: [noise()],
        pubsub: gossipsub({
          allowPublishToZeroTopicPeers: true,
          emitSelf: false,
          globalSignaturePolicy: 'StrictSign'
        })
      };

      // 添加 mDNS 发现 (如果启用)
      if (this.config.enableMDNS) {
        nodeConfig.peerDiscovery = [mdns()];
      }

      // 添加 DHT (如果启用)
      if (this.config.enableDHT) {
        nodeConfig.dht = kadDHT();
      }

      // 添加引导节点 (如果提供)
      if (this.config.bootstrapAddrs && this.config.bootstrapAddrs.length > 0) {
        if (!nodeConfig.peerDiscovery) {
          nodeConfig.peerDiscovery = [];
        }
        nodeConfig.peerDiscovery.push(bootstrap({
          list: this.config.bootstrapAddrs
        }));
      }

      // 创建 libp2p 节点
      console.log('[LIBP2P_PROVIDER] 创建 libp2p 节点，配置:', JSON.stringify(nodeConfig, (key, value) => {
        if (key === 'peerId' || key === 'pubsub') return '[Object]';
        return value;
      }, 2));

      // 创建节点
      const node = await createLibp2p(nodeConfig);
      
      // 注册自定义消息处理
      (node as any).actorMessageHandler = (event: any) => {
        console.log('[LIBP2P_PROVIDER] 收到自定义消息:', event);
        if (event && event.data) {
          this.handlePubsubMessage(event);
        }
      };

      // 当节点发现其他节点时的处理
      node.addEventListener('peer:discovery', (evt: any) => {
        const remotePeerId = evt.detail.id.toString();
        console.log(`[LIBP2P_PROVIDER] 发现新节点: ${remotePeerId}`);
        this.handlePeerDiscovery(remotePeerId);
      });

      console.log(`[LIBP2P_PROVIDER] libp2p 节点已创建`);
      return node;
    } catch (error) {
      console.error('[LIBP2P_PROVIDER] 创建 libp2p 节点失败:', error);
      throw error;
    }
  }

  /**
   * 设置订阅
   */
  private async setupSubscriptions(): Promise<void> {
    try {
      if (!this.node) {
        throw new Error('libp2p node not initialized');
      }

      // 订阅集群成员主题
      await this.node.pubsub.subscribe(this.MEMBER_TOPIC);
      console.log(`已订阅成员主题: ${this.MEMBER_TOPIC}`);

      // 订阅集群领导者主题
      await this.node.pubsub.subscribe(this.LEADER_TOPIC);
      console.log(`已订阅领导者主题: ${this.LEADER_TOPIC}`);

      // 订阅直接消息主题
      const directTopicPattern = `direct.${this.selfId}.*`;
      await this.node.pubsub.subscribe(directTopicPattern);
      console.log(`已订阅直接消息主题: ${directTopicPattern}`);

      // 设置消息处理函数
      this.node.pubsub.addEventListener('message', (event: any) => {
        this.handlePubsubMessage(event);
      });

      console.log('消息订阅和处理器已设置');
    } catch (error) {
      console.error('设置订阅失败:', error);
      throw error;
    }
  }

  /**
   * 处理pubsub消息
   * @param event 消息事件
   */
  private handlePubsubMessage(event: any): void {
    try {
      // 提取消息数据
      const data = event.data;
      const from = event.from;
      const topic = event.topic;
      
      // 检查是否有消息数据
      if (!data) {
        console.log(`[LIBP2P_PROVIDER] 收到空消息`);
        return;
      }
      
      let messageStr;
      if (data instanceof Uint8Array) {
        messageStr = new TextDecoder().decode(data);
      } else if (typeof data === 'string') {
        messageStr = data;
      } else {
        console.warn(`[LIBP2P_PROVIDER] 未知的消息格式:`, data);
        return;
      }
      
      let messageData;
      try {
        messageData = JSON.parse(messageStr);
      } catch (error) {
        console.error(`[LIBP2P_PROVIDER] 解析消息失败:`, error);
        return;
      }
      
      console.log(`[LIBP2P_PROVIDER] 收到消息: 主题=${topic}, 发送者=${from}`);
      console.log(`[LIBP2P_PROVIDER] 消息内容:`, JSON.stringify(messageData, null, 2));
      
      // 处理不同类型的消息
      if (topic.includes('actor.route')) {
        const originalMessage = messageData.message;
        
        // 处理PID地址格式的转换
        this.convertPIDAddressesToLibp2p(originalMessage);
        
        // 将消息传递给Router
        if (this.cluster && this.cluster.router) {
          console.log(`[LIBP2P_PROVIDER] 将消息传递给Router处理`);
          // @ts-ignore
          this.cluster.router.route(originalMessage);
        } else {
          console.warn(`[LIBP2P_PROVIDER] 集群或Router未初始化，无法处理消息`);
        }
      } else if (topic.includes('response')) {
        console.log(`[LIBP2P_PROVIDER] 处理响应消息`);
        
        // 提取原始响应内容和请求ID
        const requestId = topic.split('.').pop();
        const response = messageData.message;
        
        if (requestId && this.hasResponseHandler(requestId)) {
          console.log(`[LIBP2P_PROVIDER] 找到请求 ${requestId} 的响应处理程序`);
          this.handleResponse(requestId, response);
        } else {
          console.warn(`[LIBP2P_PROVIDER] 未找到请求 ${requestId} 的响应处理程序`);
        }
      }
    } catch (error) {
      console.error(`[LIBP2P_PROVIDER] 处理pubsub消息失败:`, error);
    }
  }
  
  /**
   * 递归转换消息中的PID地址为libp2p格式
   * @param obj 要处理的对象
   */
  private convertPIDAddressesToLibp2p(obj: any): void {
    if (!obj || typeof obj !== 'object') return;
    
    // 如果是PID对象，处理其地址
    if (obj.address && obj.id && typeof obj.address === 'string') {
      // 如果地址是host:port格式，且不是libp2p格式，转换它
      if (obj.address.includes(':') && !obj.address.startsWith('libp2p:')) {
        // 在这里，我们可以尝试查找对应的peerId，但简单起见先使用selfId
        const peerId = this.selfId || this.getPeerId();
        if (peerId) {
          console.log(`[LIBP2P_PROVIDER] 转换PID地址 ${obj.address} -> libp2p:${peerId}`);
          obj.address = `libp2p:${peerId}`;
        }
      }
    }
    
    // 递归处理所有属性
    for (const key in obj) {
      if (Object.prototype.hasOwnProperty.call(obj, key)) {
        const value = obj[key];
        if (value && typeof value === 'object') {
          this.convertPIDAddressesToLibp2p(value);
        }
      }
    }
    
    // 特殊处理数组
    if (Array.isArray(obj)) {
      for (let i = 0; i < obj.length; i++) {
        if (obj[i] && typeof obj[i] === 'object') {
          this.convertPIDAddressesToLibp2p(obj[i]);
        }
      }
    }
  }

  /**
   * 启动心跳计时器
   */
  private startHeartbeats(): void {
    this.startHeartbeat(); // 使用原有方法
  }

  /**
   * 设置消息响应处理器
   * @param requestId 请求ID
   * @param handler 处理函数
   */
  setResponseHandler(requestId: string, handler: (response: any) => void): void {
    console.log(`[LIBP2P_PROVIDER] 设置消息响应处理器: ${requestId}`);
    this.responseHandlers.set(requestId, handler);
  }
  
  /**
   * 移除消息响应处理器
   * @param requestId 请求ID
   */
  removeResponseHandler(requestId: string): void {
    console.log(`[LIBP2P_PROVIDER] 移除消息响应处理器: ${requestId}`);
    this.responseHandlers.delete(requestId);
  }
  
  /**
   * 检查是否有指定请求ID的响应处理器
   * @param requestId 请求ID
   * @returns 是否存在处理器
   */
  hasResponseHandler(requestId: string): boolean {
    const has = this.responseHandlers.has(requestId);
    console.log(`[LIBP2P_PROVIDER] 检查响应处理器 ${requestId}: ${has ? '存在' : '不存在'} (当前有 ${this.responseHandlers.size} 个处理器)`);
    
    if (this.responseHandlers.size > 0) {
      console.log(`[LIBP2P_PROVIDER] 当前响应处理器:`, 
        Array.from(this.responseHandlers.keys()).join(', '));
    }
    
    return has;
  }
  
  /**
   * 处理响应
   * @param requestId 请求ID
   * @param response 响应内容
   * @returns 是否成功处理
   */
  private handleResponse(requestId: string, response: any): boolean {
    console.log(`[LIBP2P_PROVIDER] 处理请求 ${requestId} 的响应:`, response);
    
    if (this.hasResponseHandler(requestId)) {
      const handler = this.responseHandlers.get(requestId);
      if (handler) {
        try {
          console.log(`[LIBP2P_PROVIDER] 调用请求 ${requestId} 的响应处理器`);
          handler(response);
          this.removeResponseHandler(requestId);
          console.log(`[LIBP2P_PROVIDER] 响应处理完成，移除处理器 ${requestId}`);
          return true;
        } catch (error) {
          console.error(`[LIBP2P_PROVIDER] 处理响应时出错:`, error);
        }
      }
    }
    
    console.warn(`[LIBP2P_PROVIDER] 未找到请求 ${requestId} 的响应处理器`);
    return false;
  }

  /**
   * 获取当前节点的peerId
   * @returns 返回peerId字符串
   */
  getPeerId(): string {
    // 优先使用selfId
    if (this.selfId) {
      return this.selfId;
    }
    
    // 如果selfId为空，尝试使用peerId对象
    if (this.peerId) {
      // 如果peerId有toString方法，调用它
      if (this.peerId.toString) {
        return this.peerId.toString();
      }
      
      // 如果没有toString方法，但有id属性
      if (this.peerId.id) {
        return this.peerId.id;
      }
    }
    
    // 如果节点已初始化，尝试从node对象获取
    if (this.node && this.node.peerId) {
      return this.node.peerId.toString();
    }
    
    // 都没有，返回空字符串
    return '';
  }
} 